package com.zyx.flinkdemo.stream.window.join;

import com.zyx.flinkdemo.pojo.ClickBean;
import com.zyx.flinkdemo.pojo.Trade;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/30 17:12
 */
public class IntervalJoinTemplate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<ClickBean> clicksData = PrepareData.getClicksData();
        DataStream<ClickBean> clickStream = env.fromCollection(clicksData)
                .assignTimestampsAndWatermarks(((WatermarkStrategy<ClickBean>) context -> new WatermarkGenerator<ClickBean>() {
                    @Override
                    public void onEvent(ClickBean clickBean, long eventTimestamp, WatermarkOutput output) {
                        output.emitWatermark(new Watermark(clickBean.getVisitTime().getTime()));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                    }
                })
                        .withTimestampAssigner((SerializableTimestampAssigner<ClickBean>) (clickBean, recordTimestamp) -> clickBean.getVisitTime().getTime()));

        List<Trade> personData = PrepareData.getTradeData();
        DataStream<Trade> tradeStream = env
                .fromCollection(personData)
                .assignTimestampsAndWatermarks(((WatermarkStrategy<Trade>) context -> new WatermarkGenerator<Trade>() {
                    @Override
                    public void onEvent(Trade trade, long eventTimestamp, WatermarkOutput output) {
                        output.emitWatermark(new Watermark(trade.getTradeTime().getTime() - 1));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                    }
                })
                        .withTimestampAssigner((SerializableTimestampAssigner<Trade>) (trade, recordTimestamp) -> trade.getTradeTime().getTime()));


        KeyedStream<ClickBean, String> clickKeyedStream = clickStream
                .keyBy((KeySelector<ClickBean, String>) ClickBean::getUser);
        KeyedStream<Trade, String> tradeKeyedStream = tradeStream
                .keyBy((KeySelector<Trade, String>) Trade::getName);

        DataStream<String> process = clickKeyedStream.intervalJoin(tradeKeyedStream)
                .between(Time.minutes(-30), Time.minutes(20))
                .process(new ProcessJoinFunction<ClickBean, Trade, String>() {
                    @Override
                    public void processElement(ClickBean left, Trade right,
                                               Context ctx, Collector<String> out) {
                        out.collect(left.toString() + " : " + right.toString());
                    }
                });

        process.print();
        env.execute("WindowIntervalJoinTumblingTemplate");
    }

}
